package servers

import (
	"encoding/json"
	"fmt"
	"github.com/dpwgc/kapokmq-go-client/conn"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	"local/config"
	"local/databases/mysql"
	"local/models"
	"local/utils"
	"strconv"
	"sync"
)

//连接的客户端,把每个客户端都放进来。Key为WsKey结构体(GroupId int64, UserId int64)。Value为websocket连接
var clients = make(map[models.WsKey]*websocket.Conn)

//广播通道，用于广播推送群聊用户发送的消息(带缓冲区，提高并发速率)
var broadcast = make(chan models.UserMessage, 10000)

var lock sync.RWMutex

// InitChat 聊天室模块初始化
func InitChat() {

	//启动广播推送协程，推送消息到各个客户端
	go pushMessages()

	//启动消息批量处理协程，借助消息队列将消息批量插入数据库
	go insertMessage()
}

// ChatLink 群聊内部消息监听
func ChatLink(c *gin.Context) {

	//获取连接的用户id与群组id
	userId := c.Param("userId")
	groupId := c.Param("groupId")

	//将id转为int64类型
	gid, _ := strconv.ParseInt(groupId, 10, 64)
	uid, _ := strconv.ParseInt(userId, 10, 64)

	//升级get请求为webSocket协议
	ws, err := config.UpGrader.Upgrade(c.Writer, c.Request, nil)

	//将当前连接的客户端放入客户端map（clients）中
	wsKey := models.WsKey{
		GroupId: gid,
		UserId:  uid,
	}
	lock.RLock()
	clients[wsKey] = ws
	lock.RUnlock()

	if err != nil {
		fmt.Println(err)
		delete(clients, wsKey) //删除map中的客户端
		return
	}
	defer ws.Close()

	for {
		//读取websocket发来的数据
		_, message, err := ws.ReadMessage()
		if err != nil {
			fmt.Println(err)
			delete(clients, wsKey) //删除map中的客户端
			break
		}

		//创建基础聊天消息模板
		chatMessage := models.Message{}
		chatMessage.MessageData = string(message)                  //聊天消息的主体（例："你好"）
		chatMessage.UserId, _ = strconv.ParseInt(userId, 10, 64)   //发送该消息的用户id
		chatMessage.GroupId, _ = strconv.ParseInt(groupId, 10, 64) //该消息所属的群组id
		chatMessage.CreateTime = utils.GetLocalDateTime()          //消息创建时间
		chatMessage.UpdateTime = utils.GetLocalDateTime()          //消息更新时间
		chatMessage.Status = 1                                     //消息状态

		//查询用户信息，获取用户头像与昵称
		user := models.User{}
		mysql.Db().Model(&user).Where("id=?", chatMessage.UserId).Find(&user)

		//创建用户聊天消息模板（往基础聊天消息模板里添加用户头像与用户昵称）
		userMessage := models.UserMessage{

			//Message中的数据
			MessageData: chatMessage.MessageData,
			UserId:      chatMessage.UserId,
			GroupId:     chatMessage.GroupId,
			CreateTime:  chatMessage.CreateTime,
			UpdateTime:  chatMessage.UpdateTime,
			Status:      chatMessage.Status,

			//在Message的基础上加上用户头像与用户昵称
			HeadImg:  user.HeadImg,
			Nickname: user.Nickname,
		}

		//如果消息为空
		if len(chatMessage.MessageData) == 0 {
			//跳过
			continue
		}

		//将聊天消息模板（用户发送的消息及用户头像昵称）插入广播频道，推送给其他websocket客户端
		broadcast <- userMessage

		//将消息发送至消息队列（流量削峰，避免数据库被冲垮）
		byteMsg, err := json.Marshal(userMessage)
		if err == nil {
			conn.ProducerSend(string(byteMsg), 0)
		}
	}
}

//广播推送消息
func pushMessages() {
	for {
		//读取通道中的消息
		msg := <-broadcast

		//轮询现有的websocket客户端
		for key, client := range clients {

			//获取该客户端的群组id（GroupId）
			gid := key.GroupId

			//匹配客户端，判断该客户端的GroupId是否与该消息的GroupId一致，如果是，则将该消息投递给该客户端
			if msg.GroupId == gid && msg.UserId != 0 && len(msg.MessageData) > 0 {
				//发送消息，含失败重试机制，重试次数:3
				for i := 0; i < 3; i++ {
					//发送消息到消费者客户端
					err := client.WriteJSON(msg)
					//如果发送成功
					if err == nil {
						//结束循环
						break
					}
					//如果到达重试次数，但仍未发送成功
					if i == 2 && err != nil {
						//客户端关闭
						client.Close()
						//删除map中的客户端
						delete(clients, key)
					}
				}
			}
		}
	}
}

//插入消息到数据库
func insertMessage() {
	for {

		//从消息队列中获取消息
		msg, _ := utils.GetMsg()

		//将userMessage转成message模板
		chatMessage := models.Message{}
		chatMessage.MessageData = msg.MessageData
		chatMessage.UserId = msg.UserId
		chatMessage.GroupId = msg.GroupId
		chatMessage.CreateTime = utils.GetLocalDateTime()
		chatMessage.UpdateTime = utils.GetLocalDateTime()
		chatMessage.Status = 1

		//开始事务
		tx := mysql.Db().Begin()

		//将message消息模板插入数据库
		err := mysql.Db().Create(&chatMessage).Error
		if err != nil {
			//错误-事务回滚
			tx.Rollback()
			continue
		}

		//更新群组信息
		group := models.Group{}
		groupText := msg.Nickname + ":" + msg.MessageData
		if len(groupText) > 16 {
			groupText = groupText[0:15]
		}
		mysql.Db().Model(&group).Where("id=?", msg.GroupId).Find(&group)
		err = mysql.Db().Model(&group).Where("id=?", msg.GroupId).Update(&models.Group{
			GroupText:  groupText,                //更新群组最新消息
			UpdateTime: utils.GetLocalDateTime(), //updateTime更新
			ChatNum:    group.ChatNum + 1,        //群组聊天消息数量+1
		}).Error
		if err != nil {
			//错误-事务回滚
			tx.Rollback()
			continue
		}

		//同步用户已读消息数量
		userGroup := models.UserGroup{}
		err = mysql.Db().Model(&userGroup).Where("user_id=? AND group_id=?", msg.UserId, msg.GroupId).Update(&models.UserGroup{
			LookNum: group.ChatNum,
		}).Error
		if err != nil {
			//错误-事务回滚
			tx.Rollback()
			continue
		}

		//事务提交
		tx.Commit()
	}
}
